package com.atguigu0.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable

/**
 * @description: sparkStreaming案例,无状态,每次统计单词的个数不会进行合并,5s内数据互不相干.
 * @time: 2020/6/15 20:54
 * @author: baojinlong
 **/
object RddQueueStreaming {
  def main(args: Array[String]): Unit = {
    // 创建SparkConf
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("wordCount")
    // 创建SteamingContext
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    // 创建rdd队列
    val rddQueue = new mutable.Queue[RDD[Int]]()
    // 创建Dstream通过rdd队列
    // val rddStream: InputDStream[Int] = ssc.queueStream(rddQueue)
    val rddStream: InputDStream[Int] = ssc.queueStream(rddQueue, oneAtATime = false)
    // 累加值
    val result: DStream[Int] = rddStream.reduce(_ + _)
    // 打印结果
    result.print
    // 开启流处理
    ssc.start()
    // 循环向rdd队列中放入rdd
    for (_ <- 1 to 5) {
      rddQueue += ssc.sparkContext.makeRDD(1 to 100, 10)
      Thread.sleep(2000)
    }
    // 等待执行
    ssc.awaitTermination()
  }
}
